package week;

import org.apache.spark.SparkConf;
import org.apache.spark.SparkContext;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.rdd.RDD;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import scala.Tuple2;

import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;

public class Work1 {


    public static void main(String[] args) {
        SparkConf sparkConf = new SparkConf();
        sparkConf.setAppName("geektimeBigDataWeek6Work1");
        sparkConf.setMaster("local[*]");

        JavaSparkContext sparkContext = new JavaSparkContext(sparkConf);

        SparkSession sparkSession = SparkSession.builder().config(sparkConf).getOrCreate();
        // test read jdbc
        String jdbcUrl = "jdbc:mysql://127.0.0.1:3306/demo2022";
        String jdbcTable = "demo_table";
        String jdbcUser = "demo";
        String jdbcPassword = "demo";
        Dataset<Row> sparkDataset = sparkSession.read().format("jdbc")
                .option("url", jdbcUrl)
                .option("dbtable", jdbcTable)
                .option("user", jdbcUser)
                .option("password", jdbcPassword)
                .load();
        System.out.println("loadData:" + sparkDataset.javaRDD().collect());

        String path = "hdfs://127.0.0.1/test2022/";
        List<String> fileNameList = new ArrayList<String>();
        fileNameList.add("1.txt");
        fileNameList.add("2.txt");
        fileNameList.add("3.txt");
        Map<String, JavaRDD<String>> fileRDDMap = new LinkedHashMap<String, JavaRDD<String>>();
        for(String fileName : fileNameList) {
            String fullPath = path + "/" + fileName;
            JavaRDD<String> fileContent = sparkContext.textFile(fullPath, 1);
            fileRDDMap.put(fileName, fileContent);
        }

        JavaRDD<Map<String, Map<String, String>>> rdd = null;
        for(String fileName : fileRDDMap.keySet()) {
            JavaRDD<String> fileContent = fileRDDMap.get(fileName);
            JavaPairRDD<String, List<String>> pairs = fileContent.mapToPair(s -> {
                String[] wordArray = s.split(" ");
                List<String> wordList = new ArrayList<>();
                for(String word : wordArray) {
                    wordList.add(word);
                }
                return new Tuple2(fileName, wordList);
            });

            JavaPairRDD<String, Integer> countRdd =  pairs.mapToPair(s ->
                 new Tuple2(s, 1)
            );

            JavaPairRDD<String, Integer> counts = countRdd.reduceByKey((a, b) -> a + b);

        }




    }
}
